package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/7
 *
 * 如何获取时间窗口的时间属性.
 *
 *  必须在函数中找TimeWindow类型的对象，才能获取时间范围。
 *
 *      reduce(ReduceFunction<T> function) ： 无法获取时间属性。
 *
 *       reduce(
 *             ReduceFunction<T> reduceFunction, ： 聚合函数
 *             WindowFunction<T, R, K, W> function
 *                  T: 输入
 *                  R： 输出
 *                  K： key的类型
 *                  W： 是窗口对象
 *            )
 *
 *       aggregate(
 *             AggregateFunction<T, ACC, V> aggFunction, WindowFunction<V, R, K, W> windowFunction)
 *
 *
 */
public class Demo8_TimeWindowGetTimeAttr
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                env
                   .socketTextStream("hadoop102", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                   .reduce(
                       new ReduceFunction<WaterSensor>()
                           {

                               @Override
                               public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                                   System.out.println("Demo6_Reduce.reduce");
                                   value2.setVc(value2.getVc() + value1.getVc());
                                   return value2;
                               }
                           },
                       new WindowFunction<WaterSensor, String, String, TimeWindow>()
                       {
                           /*
                                    窗口中每进入一条数据，就会调用一次 ReduceFunction。

                                    窗口触发运算时，此时会把 ReduceFunction聚合的结果，传递给 WindowFunction，作为它的输入！
                                    Iterable<WaterSensor> input: 当前窗口中的元素集合。
                                                        input只有一条数据，且是 ReduceFunction 聚合的最终结果
                            */
                           @Override
                           public void apply(String key, TimeWindow window, Iterable<WaterSensor> input, Collector<String> out) throws Exception {

                               System.out.println("Demo8_TimeWindowGetTimeAttr.apply");
                               WaterSensor result = input.iterator().next();
                               out.collect(key +":" + MyUtil.parseTimeWindow(window) +",聚合的最终结果:"+result);

                           }
                       })
                   .print();


          env.execute();


    }
}
